package com.hz.cheetah.server;

import com.hz.cheetah.common.bean.RpcRequest;
import com.hz.cheetah.common.bean.RpcResponse;
import com.hz.cheetah.common.util.StringUtil;
import com.hz.cheetah.server.imp.RpcSession;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

/**
 * RPC 服务端处理器（用于处理 RPC 请求）
 *
 * @author chuanyang
 * @since 1.0.0
 */
public class CheetahServerHandler extends SimpleChannelInboundHandler<RpcRequest> {

    private static final Logger LOGGER = LoggerFactory.getLogger(CheetahServerHandler.class);

    private final Map<String, Object> handlerMap;

    private MessageQueueHandler queueHandler;

    public CheetahServerHandler(Map<String, Object> handlerMap,MessageQueueHandler queueHandler) {
        this.handlerMap = handlerMap;
        this.queueHandler = queueHandler;
    }

    @Override
    public void channelRead0(final ChannelHandlerContext ctx, RpcRequest request) throws Exception {
        // 创建并初始化 RPC 响应对象
        RpcSession session = SessionsManager.resetSession(ctx.channel());
        if(request.getType() == 1){
            RpcResponse response = new RpcResponse();
            response.setRequestId(request.getRequestId());
            response.setType(1);
            ctx.writeAndFlush(response);
        }else {
            try {
                handle(session,request);
            } catch (Exception e) {
                RpcResponse response = new RpcResponse();
                response.setRequestId(request.getRequestId());
                response.setException(e);
                ctx.writeAndFlush(response);
                LOGGER.error("handle result failure:"+ctx.channel().remoteAddress().toString(), e);
            }
        }
        // 写入 RPC 响应对象并自动关闭连接
        //ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
    }

    private void handle(RpcSession rpcSession,RpcRequest request) throws Exception {
        // 获取服务对象
        String serviceName = request.getInterfaceName();
        String serviceVersion = request.getServiceVersion();
        if (StringUtil.isNotEmpty(serviceVersion)) {
            serviceName += "-" + serviceVersion;
        }
        Object serviceBean = handlerMap.get(serviceName);
        if (serviceBean == null) {
            throw new RuntimeException(String.format("can not find service bean by key: %s", serviceName));
        }

        queueHandler.submit(rpcSession,serviceBean,request);
        serviceBean = null;
        serviceName = null;
        serviceVersion = null;
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        LOGGER.error("server caught exception", cause);
        ctx.close();
        SessionsManager.sessions.remove(ctx.channel());
    }


    /**
     * Calls {@link ChannelHandlerContext#fireChannelUnregistered()} to forward
     * to the next {@link ChannelInboundHandler} in the {@link ChannelPipeline}.
     * <p>
     * Sub-classes may override this method to change behavior.
     *
     * @param ctx
     */
    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        SessionsManager.sessions.remove(ctx.channel());
        super.channelUnregistered(ctx);
    }

    /**
     * Calls {@link ChannelHandlerContext#fireChannelActive()} to forward
     * to the next {@link ChannelInboundHandler} in the {@link ChannelPipeline}.
     * <p>
     * Sub-classes may override this method to change behavior.
     *
     * @param ctx
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        SessionsManager.resetSession(ctx.channel());
        super.channelActive(ctx);
    }
}
